package com.o2o.cleaning.month.platform.ebusiness_plat.kuaishou

import java.text.SimpleDateFormat
import java.time.LocalDate
import java.util.{Calendar, Date}

import com.alibaba.fastjson.JSON
import org.apache.spark.sql.functions.lit
import com.o2o.cleaning.month.platform.ebusiness_plat.brand_modular.brand_join_res
import org.apache.spark.sql.{DataFrame, SparkSession}

object KuaiShou_Live {
  private val time: LocalDate = LocalDate.now()
  val year = time.getYear
  val month = time.getMonthValue - 1
  val platformName = "kuaishou"
  var timeStamp = getTimestamp(year, month, 30)
  val collection = if (month < 10) s"kuaishou_webcast_shop_list_210${month}" else s"kuaishou_webcast_shop_list_21${month}"
  val source_path = s"s3a://o2o-dataproces-group/zyf/livestreaming/kuaishou/sourceData/${year}/${month}/${collection}/"
  val result_path = s"s3a://o2o-dataproces-group/zyf/livestreaming/kuaishou/resultData/${year}/${month}/${collection}/"
  val result_path_mid = s"s3a://o2o-dataproces-group/zyf/livestreaming/kuaishou/resultData/${year}/${month}/mid/"
  val address_source_path = s"s3a://o2o-dimension-table/address_table/address_table_${year}/${month}/address_platform/kuaishou_address_${year}_${month}/"
  val category_path = s"s3a://o2o-dataproces-group/zyf/livestreaming/kuaishou/categoryData/2021_9/"
  val new_category_path = s"s3a://o2o-dataproces-group/zyf/livestreaming/kuaishou/categoryData/new_category_${month}/"

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName("KuaiShou_LiveStreaming")
      .config("spark.debug.maxToStringFields", "2000")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.caseSensitive", "true")
      //      .master("local[*]")
      .getOrCreate()

    val sc = spark.sparkContext
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    //            sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")
    sc.setLogLevel("ERROR")

    //计算加清洗
    val frame = Caculate(spark, source_path)
    //    val frame = Caculate(spark, "D:\\test.json")
    //              .show(10)
    frame.printSchema()
    frame.drop("add_to_field")
      .drop("expressInfo")
      .drop("serviceInfo")
      .drop("itemCategoryProps")
      .drop("commentInfo")
      .drop("shopInfo")
      .drop("youzan_source")
      .drop("mockuai_source")
      .drop("coupon")
      .drop("images")
      .drop("shopId")
      .drop("categoryId")
      .withColumnRenamed("images_new", "images")
      .withColumnRenamed("shopId_new", "shopId")
      .withColumnRenamed("categoryId_new", "categoryId")
      .withColumnRenamed("platform", "subplatformName")
      .withColumnRenamed("user_id", "userId")
      .where("sellCount > 0 and salesAmount > 0")
      .registerTempTable("t1")
    frame.repartition(30).write.orc(result_path_mid)
    println("数据中间集写出成功")
    frame.show(false)
    //    println("销售量和销售额：")
    //    spark.sql(
    //      """
    //        |select count(1) count, cast(sum(sellCount) as bigInt) sell,cast(sum(salesAmount) as decimal(20,2)) sale from t1 where sellCount > 0
    //         """.stripMargin
    //    ).show()
    val frameBrand = spark.sql(
      """
        |select *,md5(brandName) brandValueId from t1
        |""".stripMargin)
    //打标签
    val s_002 = frameBrand
      .withColumn("platformName", lit("快手"))
      .withColumn("platformId", lit("74"))
      .withColumn("shopType", lit("B"))
      .withColumn("timeStamp", lit(s"${timeStamp}"))

    println("-----关联分类-----")
    val frame1 = kuaiShouCategory(spark, s_002, category_path)
    //    frame1.show(false)
    println("-----关联地址-----")
    val frame2 = kuaiShouAddress(spark, frame1, address_source_path)
    //    frame2.repartition(30).write.orc(result_path_mid)

    //  这个是查看schema的语句
    s_002.printSchema()

    println("-----关联品牌-----")
    val brand = new brand_join_res
    brand.brandJoinResult(frame2, result_path, year, month, platformName, spark)

  }

  /**
    * 快手的事实数据和基础字段ETL，包括计算销量，价格，销售额、提取店铺URL，评论数，品牌名称
    *
    * @param spark
    * @param sourthPath
    * @return
    */
  def Caculate(spark: SparkSession, sourthPath: String): DataFrame = {
    println(sourthPath)
    // 读取原始数据
    val sourceData = spark.read.json(sourthPath).toJSON.rdd.map(line => {

      // 销量销售额计算
      val nObject = JSON.parseObject(line)
      // 获取add_to_field数组
      val flag = nObject.getOrDefault("add_to_field", "-1")
      //      println(flag)
      if (flag != "-1") {
        // 这里要过滤掉sellCount 为null 和 "" 的情况
        val array_arr = nObject.getJSONArray("add_to_field").toArray().filter(x => {
          if (!JSON.parseObject(x.toString).containsKey("sellCount")) {
            false
          } else {
            if (JSON.parseObject(x.toString).getString("sellCount") != "") {
              true
            } else {
              false
            }
          }
        })

        // 这里是处理 shopName
        var shopName = "-1"
        val flag_shopInfo = nObject.getOrDefault("shopInfo", "-1").toString
        if (flag_shopInfo != "-1") {
          val object_shopInfo = JSON.parseObject(flag_shopInfo)
          val flag_shopName = object_shopInfo.getOrDefault("shopName", "-1").toString
          if (flag_shopName != "-1") {
            shopName = flag_shopName
          } else {
            val flag_shopTitle = object_shopInfo.getOrDefault("shopTitle", "-1").toString
            if (flag_shopTitle != "-1") {
              val object_shopTitle = JSON.parseObject(flag_shopTitle)
              shopName = object_shopTitle.getOrDefault("shopName", "-1").toString
            }
          }
        } else {
          shopName = "-1"
        }
        //        println(shopName)
        // 这里是处理 categoryId 为 Double
        val flag_categoryid = nObject.getOrDefault("categoryId", "-1").toString
        var categoryId = -1
        if (flag_categoryid != "-1") {
          categoryId = flag_categoryid.toDouble.toInt
        }
        //        println(categoryId)
        // 这里是处理 shopid 带number
        var shopId = nObject.getOrDefault("shopId", "-1").toString
        if (shopId.contains("$numberLong")) {
          shopId = shopId.replace("{\"$numberLong\":\"", "")
          shopId = shopId.replace("\"}", "")
        }

        // 这里是获取 images
        val platform = nObject.getOrDefault("platform", "-1")
        var image = ""
        val flag_images = nObject.getOrDefault("images", "-1").toString
        if (platform == "xiaodian" && flag_images != "-1") {
          val images_arr = nObject.getJSONArray("images")
          if (images_arr.size() >= 1) {
            image = images_arr.get(0).toString
            if (image.contains("cdn")) {
              image = JSON.parseObject(image).get("url").toString
            }
          }
          //          val head = JSON.parseObject(images_arr.toString)
          //          image = head.getString("url")
        } else {
          image = "-1"
        }
        //        println(image)
        // 这里是获取 shopURL
        var shopInfo = ""
        var shopUrl = ""
        shopInfo = nObject.getOrDefault("shopInfo", "-1").toString
        if (shopInfo != "-1") {
          shopUrl = JSON.parseObject(shopInfo).getOrDefault("shopUrl", "-1").toString
        } else {
          shopUrl = "-1"
        }
        // 这里是获取 commentCount
        var commentInfo = ""
        var commentCount = 0
        commentInfo = nObject.getOrDefault("commentInfo", "-1").toString
        if (commentInfo != "-1") {
          commentCount = JSON.parseObject(commentInfo).getOrDefault("commentColumn", "0").toString.toInt
        } else {
          commentCount = 0
        }
        // 这里是获取品牌 Name
        var itemCategoryProps_brandName = "-1"
        var propAlias = ""
        val mockuai_source = nObject.getOrDefault("mockuai_source", "-1")
        val mokuai_brandName = if ("-1" != mockuai_source) JSON.parseObject(mockuai_source.toString).getOrDefault("brandName", "-1").toString
        else "-1"


        val itemCategoryProp = nObject.getOrDefault("itemCategoryProps", "-1")

        var itemCategoryProps: Array[AnyRef] = null
        if ("-1" != itemCategoryProp) {
          itemCategoryProps = nObject.getJSONArray("itemCategoryProps").toArray()
          for (i <- 0 to itemCategoryProps.length - 1) {
            propAlias = JSON.parseObject(itemCategoryProps(i).toString).get("propAlias").toString
            if ("品牌" == propAlias) {
              itemCategoryProps_brandName = JSON.parseObject(itemCategoryProps(i).toString).getOrDefault("propValue", "-1").toString
            }
          }
        }

        var brandName = ""
        if ("-1" != mokuai_brandName) {
          brandName = mokuai_brandName.toString
        } else if ("-1" != itemCategoryProps_brandName) {
          brandName = itemCategoryProps_brandName.toString
        } else {
          brandName = "-1"
        }
        //        println("------" + brandName)


        var sellCount = 0
        var priceText = 0.0


        /**
          * 快手算法，需要判断add_to_field ，长度为1的计算，为两个以上的用最后一个减第一个。有的里边没有sellcount ，没有的也不计算
          */
        //        try {
        // 长度为 1 的
        if (array_arr.length == 1) {
          // 暂时不做处理，sellCount默认值为0
          //                    val nObject1 = array.getJSONObject(0)
          //                    priceText = checkOutPrice(nObject1.getString("priceText"))
          //                    if (nObject1.getString("sellCount") != null && nObject1.getString("sellCount") != "") {
          //                      sellCount += checkOutSellCount(nObject1.getString("sellCount"))
          //                    }
        } else if (array_arr.length > 1) {
          // 长度大于 1 的
          /**
            * 销量用最后一天减第一天(根据时间戳降序取第一个减最后一个)
            * 价格取最低
            */
          val array_sort = array_arr.sortBy(x => {
            val crawl_date = JSON.parseObject(x.toString).get("crawl_date")
            Integer.parseInt(crawl_date.toString)
          })
          //            for (i <- 0 to array_sort.length - 1) {
          //              println("---- > " + array_sort(i).toString)
          //            }
          val lastArr = JSON.parseObject(array_sort.last.toString)
          val headArr = JSON.parseObject(array_sort.head.toString)
          //          println(lastArr.get("sellCount").toString + "111111111111111111111111")
          //          println(headArr.get("sellCount").toString + "22222222222222222222222")


          sellCount = checkOutSellCount(lastArr.getOrDefault("sellCount", "0").toString) - checkOutSellCount(headArr.getOrDefault("sellCount", "0").toString)
          // 一个月内最低的价格
          val array_price = array_arr.filter(x => {
            //              JSON.parseObject(x.toString).containsKey("priceText")
            if (!JSON.parseObject(x.toString).containsKey("priceText")) {
              false
            } else {
              if (JSON.parseObject(x.toString).getString("priceText") != "") {
                true
              } else {
                false
              }
            }
          }).sortBy(x => {
            val priceText = JSON.parseObject(x.toString).get("priceText")
            //          Integer.parseInt(priceText.toString)
            checkOutPrice(priceText.toString) * 1000.toInt
          })


          val headPriceArr = JSON.parseObject(array_price.head.toString)
          priceText = checkOutPrice(headPriceArr.getOrDefault("priceText", "0").toString)
        }
        //        } catch {
        //          case e: Exception => println(e)
        //            println(nObject.getString("good_id") + "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
        //            println(nObject.getJSONArray("add_to_field").toArray().last.toString + "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
        //            println(nObject.getJSONArray("add_to_field").toArray().head.toString + "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
        //        }

        val salesAmount = (priceText.toDouble * sellCount).toDouble.formatted("%.2f")
        //      println("good_id: "+nObject.getString("good_id")+" sellcount: "+sellCount)
        // 处理异常价格
        val flag_priceText = priceText.toInt
        if (flag_priceText == 10000
          || flag_priceText == 999999 || (flag_priceText <= 99999 && flag_priceText >= 99990) || flag_priceText == 99988 || flag_priceText == 9988
          || (flag_priceText <= 9999 && flag_priceText >= 9980) || (flag_priceText <= 999 && flag_priceText >= 990) || flag_priceText == 99979
          || flag_priceText == 9949
          || flag_priceText == 88888 || (flag_priceText <= 8889 && flag_priceText >= 8880) || flag_priceText == 8878 || flag_priceText == 888
          || flag_priceText == 6666
        ) {
          priceText = 9.9
        }
        nObject.put("brandName", brandName)
        nObject.put("shopUrl", shopUrl)
        nObject.put("images_new", image)
        nObject.put("commentCount", commentCount)
        nObject.put("shopId_new", shopId)
        nObject.put("shopName", shopName)
        nObject.put("categoryId_new", categoryId)
        nObject.put("priceText", priceText.toDouble)
        //      println("sellcount+++++++++" + sellCount)
        nObject.put("sellCount", sellCount)
        nObject.put("salesAmount", salesAmount.toDouble)
        nObject.remove("add_to_field")
      }
      nObject.toString
    }

    )
    spark.read.json(sourceData)
  }

  /**
    * 此函数用来处理销量最后带 W w 的
    *
    * @param sellcount
    * @return 处理完异常后的销量
    */
  def checkOutSellCount(sellcount: String): Int = {
    var sellCount = sellcount
    var midselld = 0.0
    var midselli = 0
    // 销量中带+号
    if (sellcount.contains('+')) {
      sellCount = sellcount.replace("+", "")
    }
    if (sellCount != null && sellCount != "") {
      if (sellCount.contains('W') || sellCount.contains('w')) {
        midselld = sellCount.dropRight(1).toDouble
        midselli = (midselld * 10000).toInt + (Math.random() * 100).toInt
      } else {
        midselli = sellCount.toInt
      }
      midselli
    } else {
      0
    }
  }


  /**
    * 此函数用来处理价格中带 -1 - , \ 的
    *
    * @param priceText
    * @return 处理完异常后的价格
    */
  def checkOutPrice(priceText: String): Double = {
    var price = priceText
    if (priceText.equals("-1")) {
      price = "0"
    } else if (priceText.contains("-")) {
      price = priceText.split("-")(0)
    } else if (priceText.contains(",")) {
      price = priceText.replace(",", "")
    } else if (priceText.contains("\"\"")) {
      price = priceText.split("\"\"")(0)
    } else {
      price
    }
    price.toDouble
  }

  /**
    * 快手的分类处理，关联firstCategoryId，secondCategoryId，thirdCategoryId，fourthCategoryId
    * 关联不到 lastcategoryid 的用上一级 thirdCategoryId 关联 fourthCategoryId 置为 ‘100xxxxxx99’,以此类推关联 secondCategoryId 和 关联firstCategoryId
    *
    * @param spark
    * @param frame
    * @param category_path
    * @return
    */
  def kuaiShouCategory(spark: SparkSession, frame: DataFrame, category_path: String): DataFrame = {
    frame.registerTempTable("t_all")
    spark.read.option("header", true).csv(category_path).registerTempTable("t_category")
    val framCategory1 = spark.sql(
      """
        |select
        |t1.*,
        |case when t1.categoryId is null then '10099' else t2.firstCategoryId end firstCategoryId,
        |case when t1.categoryId is null then '1009999' else t2.secondCategoryId end secondCategoryId,
        |case when t1.categoryId is null then '100999999' else t2.thirdCategoryId end thirdCategoryId,
        |case when t1.categoryId is null then '10099999999' else t2.fourthCategoryId end fourthCategoryId
        |from
        |t_all t1
        |left join t_category t2
        |on t1.categoryId = t2.lastcategoryid
        |where t2.lastcategoryid is not null
        |""".stripMargin)

    spark.sql(
      """
        |select
        |t1.*
        |from
        |t_all t1
        |left join t_category t2
        |on t1.categoryId = t2.lastcategoryid
        |where t2.lastcategoryid is null
        |""".stripMargin).registerTempTable("mid1")

    val framCategory2 = spark.sql(
      """
        |select
        |t1.*,
        |case when t1.categoryId is null then '10099' else t2.firstCategoryId end firstCategoryId,
        |case when t1.categoryId is null then '1009999' else t2.secondCategoryId end secondCategoryId,
        |case when t1.categoryId is null then '100999999' else t2.thirdCategoryId end thirdCategoryId,
        |case when t1.categoryId is null then '10099999999' else concat(t2.thirdCategoryId,'99') end fourthCategoryId
        |from
        |mid1 t1
        |left join t_category t2
        |on t1.categoryId = t2.thirdcategoryid
        |where t2.thirdcategoryid is not null
        |""".stripMargin)

    spark.sql(
      """
        |select
        |t1.*
        |from
        |mid1 t1
        |left join t_category t2
        |on t1.categoryId = t2.thirdcategoryid
        |where t2.thirdcategoryid is null
        |""".stripMargin).registerTempTable("mid2")

    val framCategory3 = spark.sql(
      """
        |select
        |t1.*,
        |case when t1.categoryId is null then '10099' else t2.firstCategoryId end firstCategoryId,
        |case when t1.categoryId is null then '1009999' else t2.secondCategoryId end secondCategoryId,
        |case when t1.categoryId is null then '100999999' else concat(t2.secondCategoryId,'99') end thirdCategoryId,
        |case when t1.categoryId is null then '10099999999' else concat(t2.secondCategoryId,'9999') end fourthCategoryId
        |from
        |mid2 t1
        |left join t_category t2
        |on t1.categoryId = t2.secondcategoryid
        |where t2.secondcategoryid is not null
        |""".stripMargin)

    spark.sql(
      """
        |select
        |t1.*
        |from
        |mid2 t1
        |left join t_category t2
        |on t1.categoryId = t2.secondcategoryid
        |where t2.secondcategoryid is null
        |""".stripMargin).registerTempTable("mid3")

    val framCategory4 = spark.sql(
      """
        |select
        |t1.*,
        |case when t1.categoryId is null then '10099' else t2.firstCategoryId end firstCategoryId,
        |case when t1.categoryId is null then '1009999' else concat(t2.firstcategoryId,'99') end secondCategoryId,
        |case when t1.categoryId is null then '100999999' else concat(t2.firstcategoryId,'9999') end thirdCategoryId,
        |case when t1.categoryId is null then '10099999999' else concat(t2.firstcategoryId,'999999') end fourthCategoryId
        |from
        |mid3 t1
        |left join t_category t2
        |on t1.categoryId = t2.firstcategoryId
        |""".stripMargin)

    //    println("-----没有关联上的数据(原始数据有categoryId,分类表没有)count-----")
    val framCategory5 = spark.sql(
      """
        |select
        |distinct t1.categoryId
        |from
        |mid3 t1
        |left join t_category t2
        |on t1.categoryId = t2.firstcategoryId and t1.sellCount > 0
        |where t2.firstcategoryId is null
        |""".stripMargin) //.show(false)
    framCategory5.repartition(1).write.option("header", true).csv(new_category_path)
    //    spark.sql(
    //      """
    //        |select count(1)
    //        |from t_all t1
    //        |left join t_category t2
    //        |on t1.categoryId = t2.lastcategoryid
    //        |where t2.lastcategoryid is null and t1.categoryId != '-1' and t1.sellCount > 0
    //        |""".stripMargin).show(false)
    //    spark.sql(
    //      """
    //        |select good_id,categoryId
    //        |from t_all t1
    //        |left join t_category t2
    //        |on t1.categoryId = t2.lastcategoryid
    //        |where t2.lastcategoryid is null and t1.categoryId != '-1' and t1.sellCount > 0
    //        |""".stripMargin).show(false)
    //    println("-----没有关联上的数据(原始数据没有categoryId)count-----")
    //    spark.sql(
    //      """
    //        |select count(1)
    //        |from t_all t1
    //        |left join t_category t2
    //        |on t1.categoryId = t2.lastcategoryid
    //        |where t2.lastcategoryid is null and t1.sellCount > 0
    //        |""".stripMargin).show(false)
    //    spark.sql(
    //      """
    //        |select good_id,categoryId
    //        |from t_all t1
    //        |left join t_category t2
    //        |on t1.categoryId = t2.lastcategoryid
    //        |where t2.lastcategoryid is null and t1.categoryId = '-1' and t1.sellCount > 0
    //        |""".stripMargin).show(false)
    framCategory1.unionAll(framCategory2).unionAll(framCategory3).unionAll(framCategory4)
  }

  /**
    * 快手的地址处理，关联店铺地址信息，没有关联上的默认是 0
    *
    * @param spark
    * @param frame
    * @param address_source_path
    * @return
    */
  def kuaiShouAddress(spark: SparkSession, frame: DataFrame, address_source_path: String): DataFrame = {
    frame.registerTempTable("all")
    spark.read.json(address_source_path).registerTempTable("address")
    spark.sql(
      """
        |select t1.*,
        |case when t2.shopId is not null then t2.address  else  '0' end address,
        |case when t2.shopId is not null then t2.administrative_region  else  '0' end administrative_region,
        |case when t2.shopId is not null then t2.aedzId  else  '0' end aedzId,
        |case when t2.shopId is not null then t2.city  else  '0' end city,
        |case when t2.shopId is not null then t2.city_grade  else  '0' end city_grade,
        |case when t2.shopId is not null then t2.city_origin  else  '0' end city_origin,
        |case when t2.shopId is not null then t2.district  else  '0' end district,
        |case when t2.shopId is not null then t2.district_origin  else  '0' end district_origin,
        |case when t2.shopId is not null then t2.economic_division  else  '0' end economic_division,
        |case when t2.shopId is not null then t2.if_city  else  '0' end if_city,
        |case when t2.shopId is not null then t2.if_district  else  '0' end if_district,
        |case when t2.shopId is not null then t2.if_state_level_new_areas  else  '0' end if_state_level_new_areas,
        |case when t2.shopId is not null then t2.latitude  else  '0' end latitude,
        |case when t2.shopId is not null then t2.longitude  else  '0' end longitude,
        |case when t2.shopId is not null then t2.name  else  '0' end name,
        |case when t2.shopId is not null then t2.poor_counties  else  '0' end poor_counties,
        |case when t2.shopId is not null then t2.province  else  '0' end province,
        |case when t2.shopId is not null then t2.regional_ID  else  '0' end regional_ID,
        |case when t2.shopId is not null then t2.registration_institution  else  '0' end registration_institution,
        |case when t2.shopId is not null then t2.rural_demonstration_counties  else  '0' end rural_demonstration_counties,
        |case when t2.shopId is not null then t2.rural_ecommerce  else  '0' end rural_ecommerce,
        |-- case when t2.shopId is not null then t2.shopId  else  '0' end shopId,
        |case when t2.shopId is not null then t2.the_belt_and_road_city  else  '0' end the_belt_and_road_city,
        |case when t2.shopId is not null then t2.the_belt_and_road_province  else  '0' end the_belt_and_road_province,
        |case when t2.shopId is not null then t2.the_yangtze_river_economic_zone_city  else  '0' end the_yangtze_river_economic_zone_city,
        |case when t2.shopId is not null then t2.the_yangtze_river_economic_zone_province  else  '0' end the_yangtze_river_economic_zone_province,
        |case when t2.shopId is not null then t2.town  else  '0' end town,
        |case when t2.shopId is not null then t2.urban_agglomerations  else  '0' end urban_agglomerations
        |from all t1 left join address t2
        |on t1.shopId = t2.shopId
        |""".stripMargin).drop("shopName").registerTempTable("t_all2")
    // 这一步是处理 相同的 shopId 对应不同的shopName
    spark.sql(
      """
        |select shopName,shopId from
        |(select shopName,shopId,rank() over(partition by shopId order by sum(salesAmount) desc) rank
        |from all
        |where subplatformName = 'xiaodian' group by shopName,shopId)
        |where rank = 1
        |""".stripMargin)
      .registerTempTable("t_shopName")
    val frameAddress = spark.sql(
      """
        |select t1.*,t2.shopName
        |from t_all2 t1
        |left join t_shopName t2
        |on t1.shopId = t2.shopId
        |""".stripMargin)
    frameAddress
  }

  /**
    * 获得前几天的时间戳
    *
    * @param month
    * @param day
    * @return
    */
  def getTimestamp(year: Int, month: Int, day: Int): Long = {
    val format = new SimpleDateFormat("yyyy-MM-dd");
    val calendar: Calendar = Calendar.getInstance()
    calendar.set(year, month - 1, day)
    val time: Date = calendar.getTime
    val qyt = format.format(time)
    val timeStamp: Long = format.parse(qyt).getTime
    val result = timeStamp.toString.substring(0, timeStamp.toString.length - 3).toLong
    result
  }
}
